package com.enjoy.redis.mq;

import org.junit.Test;
import redis.clients.jedis.*;
import redis.clients.jedis.params.XReadGroupParams;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class StreamTest {

    public static final String STREAM_KEY = "streamKey";

    public static final String GROUP_A = "group:a";

    @Test
    public void produce() {
        Jedis jedis = new Jedis(new HostAndPort("192.168.126.128", 6379));

        // 消息
        Map<String, String> map = new HashMap<>();
        map.put("name", "aa");
        map.put("age", "11");

        // 发送
        StreamEntryID streamEntryID = jedis.xadd(STREAM_KEY, StreamEntryID.NEW_ENTRY, map);

        System.out.println("发布消息到redisMq，返回消息id=" + streamEntryID.toString());

        // 关于消息信息
        StreamInfo streamInfo = jedis.xinfoStream(STREAM_KEY);
        System.out.println("streamInfo.getStreamInfo() = " + streamInfo.getStreamInfo());
//        System.out.println("streamInfo.getLength() = " + streamInfo.getLength());
//        System.out.println("streamInfo.getRadixTreeKeys() = " + streamInfo.getRadixTreeKeys());
//        System.out.println("streamInfo.getRadixTreeNodes() = " + streamInfo.getRadixTreeNodes());
//        System.out.println("streamInfo.getGroups() = " + streamInfo.getGroups());
//        System.out.println("streamInfo.getLastGeneratedId() = " + streamInfo.getLastGeneratedId());
//        System.out.println("streamInfo.getFirstEntry() = " + streamInfo.getFirstEntry());
//        System.out.println("streamInfo.getLastEntry() = " + streamInfo.getLastEntry());

        // 获取消费群组
        List<StreamGroupInfo> streamGroupInfos = jedis.xinfoGroup(STREAM_KEY);
        for (StreamGroupInfo streamGroupInfo : streamGroupInfos) {
            System.out.println("streamGroupInfo.getGroupInfo() = " + streamGroupInfo.getGroupInfo());
//            System.out.println("streamGroupInfo.getName() = " + streamGroupInfo.getName());
//            System.out.println("streamGroupInfo.getConsumers() = " + streamGroupInfo.getConsumers());
//            System.out.println("streamGroupInfo.getPending() = " + streamGroupInfo.getPending());
//            System.out.println("streamGroupInfo.getLastDeliveredId() = " + streamGroupInfo.getLastDeliveredId());
        }
    }

    @Test
    public void consumer() {
        Jedis jedis = new Jedis(new HostAndPort("192.168.126.128", 6379));
//        List<StreamGroupInfo> streamGroupInfos = jedis.xinfoGroup(STREAM_KEY);
//        for (StreamGroupInfo groupInfo : streamGroupInfos) {
//            System.out.println("groupInfo.getName() = " + groupInfo.getName());
//        }
        // 0-0表示从头开始消费消息
        StreamEntryID streamEntryID = new StreamEntryID("0-0");
        // makeStream表示没有时是否自动创建stream,但是如果有，再创建会自动异常
        jedis.xgroupCreate(STREAM_KEY, GROUP_A, streamEntryID, false);

        // 阻塞式获取2条消息
        XReadGroupParams xReadGroupParams = new XReadGroupParams().block(0).count(2);

        Map<String, StreamEntryID> streams = new HashMap<>();
        streams.put(STREAM_KEY, StreamEntryID.UNRECEIVED_ENTRY);

        List<Map.Entry<String, List<StreamEntry>>> result = jedis.xreadGroup(GROUP_A, "consumer:a", xReadGroupParams, streams);
        System.out.println("result = " + result);

        // 消费者
        List<StreamConsumersInfo> consumersInfos = jedis.xinfoConsumers(STREAM_KEY, GROUP_A);
        for (StreamConsumersInfo consumersInfo : consumersInfos) {
            System.out.println("consumersInfo.getConsumerInfo() = " + consumersInfo.getConsumerInfo());
//            System.out.println("consumersInfo.getName() = " + consumersInfo.getName());
//            System.out.println("consumersInfo.getPending() = " + consumersInfo.getPending());
//            System.out.println("consumersInfo.getIdle() = " + consumersInfo.getIdle());
        }

        // 消费群组
        List<StreamGroupInfo> streamGroupInfos = jedis.xinfoGroup(STREAM_KEY);
        for (StreamGroupInfo streamGroupInfo : streamGroupInfos) {
            System.out.println("streamGroupInfo.getGroupInfo() = " + streamGroupInfo.getGroupInfo());
        }

        for (Map.Entry<String, List<StreamEntry>> entry : result) {
            for (StreamEntry streamEntry : entry.getValue()) {

                jedis.xack(STREAM_KEY, GROUP_A,streamEntry.getID());

                List<StreamConsumersInfo> consumers = jedis.xinfoConsumers(STREAM_KEY, GROUP_A);
                for (StreamConsumersInfo consumer : consumers) {
                    System.out.println("consumer.getConsumerInfo() = " + consumer.getConsumerInfo());
                }
                List<StreamGroupInfo> streamGroup = jedis.xinfoGroup(STREAM_KEY);

                for (StreamGroupInfo streamGroupInfo : streamGroup) {
                    System.out.println("streamGroupInfo.getGroupInfo() = " + streamGroupInfo.getGroupInfo());
                }
            }
        }
    }

}
